#pragma once
#include <memory>
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "../comm/Log.h"
#include "../comm/util.h"
#include "../comm/threadpool.h"
#include "../comm/msg.pb.h"
#include "../comm/mq.pb.h"
#include "consumer.h"
#include "route.h"
#include "virtualhost.h"

namespace ns_channel
{
    using namespace ns_log;
    using namespace ns_util;
    using namespace ns_proto;
    using namespace ns_consume;
    using namespace ns_virtual;
    using namespace ns_route;

    using ProtobufCodecPtr = shared_ptr<ProtobufCodec>;

    // 信道中的函数参数采用协议的方式接收
    using openChannelRequestPtr = shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = shared_ptr<basicCancelRequest>;

    // 信道所提供的服务操作
    class Channel
    {
    public:
        using ptr = shared_ptr<Channel>;

        Channel(const string &cid, shared_ptr<ConsumerManage> &cmp,
                const muduo::net::TcpConnectionPtr &conn,
                const ProtobufCodecPtr &codec,
                const shared_ptr<VirtualHost> &host,
                const shared_ptr<ThreadPool> &pool)
            : _cid(cid), _cmp(cmp), _conn(conn), _codec(codec), _host(host), _pool(pool) {}

        ~Channel()
        {
            if (_consumer.get() != nullptr)
            {
                _cmp->delete_queue_consumer(_consumer->_qname, _consumer->_tag);
            }
        }

        // 以下针对虚拟机模块调用队列执行操作时，同时还需要注意处理消费者
        // 声明、删除交换机
        void declare_exchange(const declareExchangeRequestPtr &req)
        {
            _host->declare_exchange(req->exchange_name(), req->exchange_type(),
                                    req->durable(), req->auto_delete(), req->args());
            // 构建相应数据并发送
            buildResponseSend(req->rid(), req->cid(), true);
        }
        void delete_exchange(const deleteExchangeRequestPtr &req)
        {
            // 同时内部还删除了交换机的绑定信息
            _host->delete_exchange(req->exchange_name());

            buildResponseSend(req->rid(), req->cid(), true);
        }

        // 声明、删除队列
        void declare_queue(const declareQueueRequestPtr &req)
        {
            // 同时内部还创建了队列消息文件
            _host->declare_queue(req->queue_name(), req->durable(), req->exclusive(),
                                 req->auto_delete(), req->args());
            // 还要初始化队列消费者管理结构
            _cmp->init_queue_consumer(req->queue_name());

            buildResponseSend(req->rid(), req->cid(), true);
        }
        void delete_queue(const deleteQueueRequestPtr &req)
        {
            // 同时内部删除了队列消息文件+绑定信息
            _host->delete_queue(req->queue_name());
            // 还要删除订阅队列的所有消费者
            _cmp->destroy_queue_consumer(req->queue_name());

            buildResponseSend(req->rid(), req->cid(), true);
        }

        // 绑定、解绑交换机与队列信息
        void bind(const queueBindRequestPtr &req)
        {
            _host->declare_bind(req->exchange_name(),
                                req->queue_name(), req->binding_key());

            buildResponseSend(req->rid(), req->cid(), true);
        }
        void unbind(const queueUnBindRequestPtr &req)
        {
            _host->delete_bind(req->exchange_name(), req->queue_name());

            buildResponseSend(req->rid(), req->cid(), true);
        }

        // 发布消息(给交换机去路由选择队列)-"生产者"
        void basic_publish(const basicPublishRequestPtr &req)
        {
            // 1.进行路由判断交换机下的所有队列
            // 1.1获取指定交换机
            Exchange::ptr ep = _host->select_exchange(req->exchange_name());
            if (ep.get() == nullptr)
            {
                LOG(WARNING) << "发布消息失败，该交换机不存在" << endl;
                return buildResponseSend(req->rid(), req->cid(), false);
            }
            // 1.2获取交换机的所有绑定信息
            QueueBindMapPtr qbindp = _host->exchange_binds(req->exchange_name());
            if (qbindp.get() == nullptr)
            {
                LOG(WARNING) << "发布消息失败，该交换机下不存在队列绑定信息" << endl;
                return buildResponseSend(req->rid(), req->cid(), false);
            }
            // 1.3轮训route所有队列绑定信息的binding_key
            auto binding_pair = qbindp->begin();
            for (; binding_pair != qbindp->end(); binding_pair++)
            {
                // 路由需要交换机类型
                string routing_key;
                if (req->has_attributes()) // 防止消息描述信息不存在
                {
                    routing_key = req->attributes().routing_key();
                }
                bool ret = Router::route(ep->_type, routing_key,
                                         binding_pair->second->_binding_key);
                if (ret)
                {
                    // 2.信息添加进能够成功路由的队列中
                    _host->insert_msg(binding_pair->first,
                                      req->mutable_attributes(), req->body());

                    // 3.向线程池中添加任务(向队列的订阅者推送消息)
                    _pool->Push(std::bind(&Channel::consume, this, binding_pair->first));
                }
            }

            buildResponseSend(req->rid(), req->cid(), true);
        }

        // 消息确认(删除)
        void basic_ack(const basicAckRequestPtr &req)
        {
            _host->comfirm_msg(req->queue_name(), req->message_id());

            buildResponseSend(req->rid(), req->cid(), true);
        }

        // 订阅队列消息-"消费者"
        void basic_consum(const basicConsumeRequestPtr &req)
        {
            // 1.判断队列是否存在
            if (!_host->exist_queue(req->queue_name()))
            {
                LOG(WARNING) << "订阅队列消息失败，队列不存在" << endl;
                buildResponseSend(req->rid(), req->cid(), false);
            }

            // 2.创建队列消费者(callback功能是进行消息推送任务)
            auto cb = std::bind(&Channel::callback, this, placeholders::_1,
                                placeholders::_2, placeholders::_3);
            _consumer = _cmp->insert_queue_consumer(
                req->queue_name(), req->consumer_tag(), req->auto_ack(), cb);

            buildResponseSend(req->rid(), req->cid(), true);
        }

        // 队列消息取消订阅(释放消费者)
        void basic_cancel(const basicCancelRequestPtr &req)
        {
            _cmp->delete_queue_consumer(req->queue_name(), req->consumer_tag());

            buildResponseSend(req->rid(), req->cid(), true);
        }

    private:
        void buildResponseSend(const string &rid, const string &cid, const bool &ok)
        {
            // 构建响应
            basicCommonResponse rsp;
            rsp.set_rid(rid);
            rsp.set_cid(cid);
            rsp.set_ok(ok); // 取决于请求执行成功与否
            _codec->send(_conn, rsp);
        }

        // 推送消息给订阅消费者，然后让消费者去处理队列消息
        void consume(const string &qname)
        {
            // 1.取出队列消息
            shared_ptr<Message> mp = _host->que_front_msg(qname);
            if (mp.get() == nullptr)
            {
                LOG(WARNING) << "队列消息执行失败，该队列下没有消息" << endl;
                return;
            }

            // 2.轮训选择消费者
            Consumer::ptr cp = _cmp->select_queue_consumer(qname);
            if (mp.get() == nullptr)
            {
                LOG(WARNING) << "队列消息执行失败，该队列下不存在消费者" << endl;
                return;
            }

            // 3.调用消费者回调函数进行消息的推送
            cp->_callback(cp->_tag, mp->mutable_payload()->mutable_attributes(),
                          mp->payload().body());
            // 4.消息确认
            if (cp->_auto_ack == true)
            {
                _host->comfirm_msg(qname, mp->payload().attributes().id());
            }
            // 否则就等到收到确认消息之后进行确认
        }

        // 设置消费者的回调函数
        void callback(const std::string &tag, const ns_proto::Attributes *attributes,
                      const std::string &body)
        {
            // 通过参数组织出推送消息的请求 推送给信道channel对应的客户端
            basicConsumeResponse rsp;
            rsp.set_cid(_cid);
            rsp.set_consumer_tag(tag);
            rsp.set_body(body);
            if (attributes != nullptr)
            {
                rsp.mutable_attributes()->set_id(attributes->id());
                rsp.mutable_attributes()->set_routing_key(attributes->routing_key());
                rsp.mutable_attributes()->set_del_mod(attributes->del_mod());
            }

            _codec->send(_conn, rsp);
        }

    private:
        // 整合(虚拟机+消费者+线程池)搭配网络+协议实现信道的操作
        // 用于给建立好连接的客户端提供服务

        string _cid;                        // 信道标识id
        Consumer::ptr _consumer;            // 信道对应的消费者
        shared_ptr<ConsumerManage> _cmp;    // 消费者管理
        muduo::net::TcpConnectionPtr _conn; // 通信连接(数据发送)
        ProtobufCodecPtr _codec;            // 协议处理器(没必要重复定义，共用)
        shared_ptr<VirtualHost> _host;      // 虚拟机管理句柄
        shared_ptr<ThreadPool> _pool;       // 线程池(用于数据传输给消费者)
    };

    // 信道的管理
    class ChannelManage
    {
    public:
        using Ptr = shared_ptr<ChannelManage>;

        ChannelManage() {}

        void createChannel(const string &cid, shared_ptr<ConsumerManage> &cmp,
                           const muduo::net::TcpConnectionPtr &conn,
                           const ProtobufCodecPtr &codec,
                           const shared_ptr<VirtualHost> &host,
                           const shared_ptr<ThreadPool> &pool)
        {
            if (_channels.count(cid))
            {
                LOG(WARNING) << "创建信道失败,该信道已经存在" << endl;
                return;
            }
            Channel::ptr cp = make_shared<Channel>(cid, cmp, conn, codec, host, pool);
            _channels.insert(make_pair(cid, cp));
        }

        void closeChannel(const string &cid)
        {
            if (!_channels.count(cid))
            {
                LOG(WARNING) << "关闭信道失败，不存在该id的信道" << endl;
                return;
            }
            _channels.erase(cid);
        }

        Channel::ptr getChannel(const string &cid)
        {
            if (!_channels.count(cid))
            {
                LOG(WARNING) << "获取信道失败，不存在该id的信道" << endl;
                return Channel::ptr();
            }
            return _channels[cid];
        }

    private:
        mutex _mtx;
        unordered_map<string, Channel::ptr> _channels;
    };

}